package tom.spring.cloud.stream.receiver;

import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.utils.DateUtils;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.annotation.StreamMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import tom.spring.cloud.msg.MyInput;
import tom.spring.cloud.msg.MyOutput;

import java.util.Calendar;
import java.util.Date;

/**
 * Spring-cloud-stream注解，都能实现管道监听，功能更强大一些.
 * StreamListener默认使用MappingJackson2MessageConverter做消息转换
 * 如果多个StreamListener监听同一个管道，则不能有返回值
 * @author ZHUFEIFEI
 */
@Slf4j
@EnableBinding({MyInput.class})
public class Consumer {

    /**
     * 接收消息并做出反馈，发送方需要监听OUTPUT通道
     *
     * 消费者，INPUT消费对应通道：input_1
     *        OUTPUT反馈对应通道: output_1
     *
     *        见配置文件application-consumer.yml
     *
     * @param date
     * @return
     */
    @StreamListener(value = MyInput.INPUT)
    @SendTo(MyOutput.OUTPUT)
    public String receive010(@Payload Date date) {
        log.info("receive010 date: {}", date);
        return "received-" + DateUtils.formatDate(date, "HH:mm:ss");
    }

    /**
     * 自定义消息转换，需要注入到容器并且标注StreamMessageConverter
     * @return
     */
    @Bean
    @StreamMessageConverter
    public MessageConverter myMessageConvert() {
        return new MyMessageConvert();
    }

    /**
     * 自定义消息转换器，默认提供了一些转换器可以完成json和基本类型的转换，
     * 如特殊需求可以自定义转换器转换
     */
    @Slf4j
    static class MyMessageConvert implements MessageConverter {

        /**
         * 收到消息时会执行该方法转换
         * @param message
         * @param targetClass
         * @return
         */
        @Override
        public Object fromMessage(Message<?> message, Class<?> targetClass) {
            log.info("fromMessage ==> {} {}", message, targetClass);
            if (message.getPayload() instanceof byte[]) {
                String str = new String((byte[])message.getPayload());
                log.info("fromMessage ==> str {}", str);
                //str带双引号需要去除掉
                Date d = DateUtils.parseDate(str.replaceAll("\"", ""), new String[]{"yyyy-MM-dd'T'HH:mm:ss.SSSZ"});
                if (d != null) {
                    log.info("fromMessage ==> parse {} to {}", str, d.getTime());
                    Calendar c = Calendar.getInstance();
                    c.setTime(d);
                    c.add(Calendar.HOUR, 8);
                    return c.getTime();
                }
            }
            //返回null将执行下一个转换器，看CompositeMessageConvert代码
            return null;
        }

        /**
         * 发出消息时会先执行该方法转换消息
         * @param payload
         * @param headers
         * @return
         */
        @Override
        public Message<?> toMessage(Object payload, MessageHeaders headers) {
            log.info("toMessage ==> {} {}", payload, headers);
            String str = payload + " converted!";
            return MessageBuilder.createMessage(str, headers);
        }
    }
}
